from confluent_kafka import Producer
import json
from config.config import settings


class QueuePush:
    def __init__(self,message_data):
        self.p = Producer({'bootstrap.servers': settings.queue_push_sever})
        self.room_id = message_data["room_id"]
        self.productDetails = message_data["ProductDetails"]
        self.user_id = message_data["user_id"]
        self.user_name = message_data["user_name"]
        self.event_time = message_data["event_time"]
        self.user_question = message_data["user_question"]
        self.intelligent_reply = message_data["intelligent_reply"]

    def delivery_report(self, err, msg):
        if err is not None:
            print('Message delivery failed: {}'.format(err))
        else:
            print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition()))

    def push(self):
        kafka_data = {
            "room_id": self.room_id,
            "ProductDetails": self.productDetails,
            "user_id": self.user_id,
            "user_name": self.user_name,
            "event_time": self.event_time,
            "user_question": self.user_question,
            "intelligent_reply":self.intelligent_reply
        }
        self.p.produce(settings.generate_reply_topic, key=str(self.room_id), value=json.dumps(kafka_data).encode('utf-8'),
                       callback=self.delivery_report)
        self.p.poll(0)
        self.p.flush()  # 确保所有消息都被发送
        print("成功推送消息采集队列")
